package com.offcn.bigdata.spark.streaming.p2

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * window窗口操作
  *     窗口操作，其实是一个局部时间段内数据的统计，
  *     人话：每隔多长时间(M)统计过去多长时间(N)内产生的数据。
  *     M在这里我们称之为滑动长度：slideInterval
  *     N在这里我们称之为窗口长度：windowLength
  *  需要注意的地方：
  *     1、滑动的频率和窗口的长度应该是批次间隔时间整数倍。
  *     2、因为滑动窗口的原因，所以我们内存需要足够大，以便于能够存放的下一个窗口中的数据
  *  绝大多数的宽依赖算子都会有这个window窗口操作，比如reduceByKeyAndWindow，groupByKeyAndWindow等等
  *
  *  需求：每隔2个时间单位，统计过去3个时间单位内所产生的数据
  */
object _02StreamingWindowOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName("_02StreamingWindowOps")
            .set("spark.streaming.kafka.maxRatePerPartition", "10")
        val batchInterval = 2
        val ssc = new StreamingContext(conf, Seconds(batchInterval))

        val lines = ssc.socketTextStream("bigdata01", 9999)
        val windowLength = Seconds(batchInterval * 3)
        val slideInterval = Seconds(batchInterval * 2)
        val ret = lines.flatMap(_.split("\\s+")).map((_, 1))
                .reduceByKeyAndWindow(
                        (v1:Int, v2:Int) => v1 + v2,
                        windowLength,
                        slideInterval
                    )

        ret.print()

        ssc.start()
        ssc.awaitTermination()
    }
}
